use std::cmp::Ordering;

use crate::block::BlockContents;
use crate::options::Options;

use integer_encoding::{FixedIntWriter, VarIntWriter};

/// BlockBuilder contains functionality for building a block consisting of consecutive key-value entries.
///
/// 这段 Rust 代码定义了一个名为 BlockBuilder 的结构体，从注释和字段来看，
/// 它的作用是构建由连续键值对（key-value entries）组成的数据块（block），这通常出现在需要高效存储和检索键值对的系统中（例如数据库、存储引擎等）。
///
/// 核心设计思路：
///
/// 在存储连续键值对时，为了节省空间，通常会采用键的增量编码（例如只存储当前键与前一个键的差异部分）。但这种方式会导致查询时需要从块的开头逐个解析，效率较低。
///
/// 因此引入了 “重启点” 机制：
///
/// 每隔一定数量的键值对（由配置决定）设置一个重启点
///
/// 重启点处的键会存储完整值（而非增量差异）
///
/// 查询时可以先定位到最近的重启点，再从该点开始解析，减少无效计算
///
/// BlockBuilder 的工作流程大致是：
///
/// 接收键值对，通过增量编码将其写入 buffer
///
/// 定期记录重启点到 restarts（根据 restart_counter 判断）
///
/// 最终生成一个包含所有键值对数据和重启点信息的完整块
///
/// 这种结构在 LevelDB、RocksDB 等嵌入式数据库的存储引擎中非常常见，用于平衡存储空间和查询性能。
pub struct BlockBuilder {
    /// 存储配置选项，可能包含块大小限制、重启点间隔等参数。
    opt: Options,
    /// 用于存储编码后的键值对数据，最终会构成块的主体内容。
    buffer: Vec<u8>,
    /// 存储 “重启点”（restart points）的偏移量，重启点是为了优化查询性能而设计的（下文解释）。
    restarts: Vec<u32>,

    /// 记录上一个插入的键，用于实现键的增量编码（减少重复存储）。
    last_key: Vec<u8>,
    /// 计数当前重启点区间内的键值对数量，达到阈值后会触发新的重启点。
    restart_counter: usize,
    /// 记录当前块中已插入的键值对总数。
    counter: usize,
}

impl BlockBuilder {
    pub fn new(o: Options) -> BlockBuilder {
        let mut restarts = vec![0];
        restarts.reserve(1023);

        BlockBuilder {
            buffer: Vec::with_capacity(o.block_size),
            opt: o,
            restarts,
            last_key: Vec::new(),
            restart_counter: 0,
            counter: 0,
        }
    }

    pub fn entries(&self) -> usize {
        self.counter
    }

    pub fn last_key(&self) -> &[u8] {
        &self.last_key
    }

    pub fn size_estimate(&self) -> usize {
        self.buffer.len() + 4 * self.restarts.len() + 4
    }

    pub fn reset(&mut self) {
        self.buffer.clear();
        self.restarts.clear();
        self.last_key.clear();
        self.restart_counter = 0;
        self.counter = 0;
    }

    /// 这段代码是 BlockBuilder 结构体的 add 方法，用于向数据块中添加键值对（key-value），核心功能是高效地编码并存储键值对，同时通过 "增量编码" 和 "重启点" 机制优化存储空间和查询性能。
    pub fn add(&mut self, key: &[u8], val: &[u8]) {
        // 前置检查：
        // 确保重启点计数器未超过配置的间隔（block_restart_interval）
        // 确保新添加的键大于上一个键（维持键的有序性，这是很多存储引擎的要求）
        assert!(self.restart_counter <= self.opt.block_restart_interval);
        assert!(
            self.buffer.is_empty()
                || self.opt.cmp.cmp(self.last_key.as_slice(), key) == Ordering::Less
        );

        let mut shared = 0;
        // 如果当前未达到重启点间隔，计算新键与上一个键的共享前缀长度（shared）
        // 例如：上一个键是 "apple"，新键是 "apples"，则共享前缀长度为 5（"apple"）
        if self.restart_counter < self.opt.block_restart_interval {
            let smallest = if self.last_key.len() < key.len() {
                self.last_key.len()
            } else {
                key.len()
            };

            while shared < smallest && self.last_key[shared] == key[shared] {
                shared += 1;
            }
        } else {
            // 当达到配置的重启点间隔时，记录当前缓冲区长度作为新的重启点
            // 清空上一个键的记录，重置重启点计数器
            // 重启点处的键会完整存储（不使用增量编码），方便后续快速查询
            self.restarts.push(self.buffer.len() as u32);
            self.last_key.clear();
            self.restart_counter = 0;
        }

        let non_shared = key.len() - shared;
        // 用可变长度编码（varint）写入：共享前缀长度、非共享部分长度、值长度
        // 写入键的非共享部分（只存差异，节省空间）
        // 写入完整的值
        self.buffer.write_varint(shared).expect("write to buffer failed");
        self.buffer.write_varint(non_shared).expect("write to buffer failed");
        self.buffer.write_varint(val.len()).expect("write to buffer failed");
        self.buffer.extend_from_slice(&key[shared..]);
        self.buffer.extend_from_slice(val);

        // Update key
        // 更新上一个键的记录（只保留共享前缀 + 新的非共享部分）
        // 递增重启点计数器和总键值对计数器
        self.last_key.resize(shared, 0);
        self.last_key.extend_from_slice(&key[shared..]);

        self.restart_counter += 1;
        self.counter += 1;
    }

    /// 这段代码是 BlockBuilder 结构体的 finish 方法，用于完成数据块（block）的构建并生成最终的块内容。
    /// 它会对之前通过 add 方法添加的所有键值对数据进行收尾处理，生成一个可以被持久化存储或传输的完整数据块。
    ///
    /// 经过 finish 处理后，生成的块内容在内存中是这样的结构：
    ///
    /// \[键值对数据区\]\[重启点偏移量列表\]\[重启点数量\]
    ///
    /// 键值对数据区：通过 add 方法写入的所有键值对（使用增量编码）
    ///
    /// 重启点偏移量列表：每个重启点在数据区中的位置（固定长度编码）
    ///
    /// 重启点数量：标识有多少个重启点（固定长度编码）
    ///
    /// 这个方法的作用是将零散的键值对数据和重启点信息整合为一个标准格式的完整数据块，使其可以被正确地存储和后续读取解析。
    /// 这是存储引擎中数据块构建的最后一步，常见于 LevelDB、RocksDB 等键值存储系统的实现中。
    pub fn finish(mut self) -> BlockContents {
        // 首先为缓冲区（buffer）预留足够空间，用于存储重启点（restarts）数据
        // 预留的空间大小为：重启点数量 ×4 字节（每个重启点是 u32 类型） + 4 字节（用于存储重启点总数）
        self.buffer.reserve(self.restarts.len() * 4 + 4);

        // 1. Append RESTARTS
        // 遍历 restarts 数组，将每个重启点的偏移量（u32 类型）以固定长度编码写入缓冲区
        // 这些重启点记录了块中各个 "完整键" 的位置，用于后续查询时快速定位
        for r in self.restarts.iter() {
            self.buffer.write_fixedint(*r).expect("write to buffer failed");
        }

        // 2. Append N_RESTARTS
        // 在所有重启点之后，写入一个固定长度的整数（u32），表示重启点的总数量
        // 这使得读取方能够知道有多少个重启点需要处理
        self.buffer.write_fixedint(self.restarts.len() as u32).expect("write to buffer failed");

        // done
        // 最终返回构建完成的缓冲区（self.buffer），作为完整的块内容（BlockContents）
        self.buffer.into()
    }
}

#[cfg(test)]
mod tests {
    use super::*;
    use crate::options;

    fn get_data() -> Vec<(&'static [u8], &'static [u8])> {
        vec![
            ("key1".as_bytes(), "value1".as_bytes()),
            (
                "loooooooooooooooooooooooooooooooooongerkey1".as_bytes(),
                "shrtvl1".as_bytes(),
            ),
            ("medium length key 1".as_bytes(), "some value 2".as_bytes()),
            ("prefix_key1".as_bytes(), "value".as_bytes()),
            ("prefix_key2".as_bytes(), "value".as_bytes()),
            ("prefix_key3".as_bytes(), "value".as_bytes()),
        ]
    }

    #[test]
    fn test_block_builder_sanity() {
        let mut o = options::for_test();
        o.block_restart_interval = 3;
        let mut builder = BlockBuilder::new(o);
        let d = get_data();

        for &(k, v) in d.iter() {
            builder.add(k, v);
            assert!(builder.restart_counter <= 3);
            assert_eq!(builder.last_key(), k);
        }

        assert_eq!(149, builder.size_estimate());
        assert_eq!(d.len(), builder.entries());

        let block = builder.finish();
        assert_eq!(block.len(), 149);
    }

    #[test]
    fn test_block_builder_reset() {
        let mut o = options::for_test();
        o.block_restart_interval = 3;
        let mut builder = BlockBuilder::new(o);
        let d = get_data();

        for &(k, v) in d.iter() {
            builder.add(k, v);
            assert!(builder.restart_counter <= 3);
            assert_eq!(builder.last_key(), k);
        }

        assert_eq!(d.len(), builder.entries());
        builder.reset();
        assert_eq!(0, builder.entries());
        assert_eq!(4, builder.size_estimate());
    }

    #[test]
    #[should_panic]
    fn test_block_builder_panics() {
        let mut d = get_data();
        // Identical key as d[3].
        d[4].0 = b"prefix_key1";

        let mut builder = BlockBuilder::new(options::for_test());
        for &(k, v) in d.iter() {
            builder.add(k, v);
            assert_eq!(k, builder.last_key());
        }
    }
    // Additional test coverage is provided by tests in block.rs.
}
